package com.ysstech.kettle.service.impl;

import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.ysstech.common.enums.YssEnum;
import com.ysstech.common.util.CollectionUtil;
import com.ysstech.common.util.DateUtil;
import com.ysstech.common.util.JsonUtil;
import com.ysstech.common.util.StringUtil;
import com.ysstech.common.util.fileutil.FileUtil;
import com.ysstech.kettle.entity.JobParam;
import com.ysstech.kettle.entity.JobRecord;
import com.ysstech.kettle.enums.KettleDbDriverEnum;
import com.ysstech.kettle.enums.KettleDbUrlEnum;
import com.ysstech.kettle.enums.RunResultEnum;
import com.ysstech.kettle.init.KettleConfig;
import com.ysstech.kettle.service.JobRecordService;
import com.ysstech.kettle.util.KettleLogUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.lang.StringUtils;
import org.pentaho.di.core.ProgressNullMonitorListener;
import org.pentaho.di.core.logging.LogLevel;
import org.pentaho.di.job.Job;
import org.pentaho.di.job.JobMeta;
import org.pentaho.di.repository.AbstractRepository;
import org.pentaho.di.repository.RepositoryDirectoryInterface;
import org.springframework.util.CollectionUtils;

import java.math.BigDecimal;
import java.time.LocalDateTime;
import java.time.temporal.ChronoUnit;
import java.util.*;
import java.util.concurrent.Callable;

@Slf4j
public class KtttleExecuteThread implements Callable<Boolean> {
    private KettleConfig kettleConfig;
    private JobRecordService jobRecordService;
    private Boolean isTaskDetail;
    private Boolean isTask;
    private String taskId;
    private List<JobParam> listParam;

    public KtttleExecuteThread(JobRecordService jobRecordService, KettleConfig kettleConfig, List<JobParam> listParam, Boolean isTaskDetail, Boolean isTask, String taskId) {
        this.isTaskDetail = isTaskDetail;
        this.isTask = isTask;
        this.taskId = taskId;
        this.kettleConfig = kettleConfig;
        this.listParam = listParam;
        this.jobRecordService = jobRecordService;
    }

    @Override
    public Boolean call() throws Exception {
        // 设置执行参数
        Map<String, Object> params = new HashMap<>();
        params.put("IS_TASK_DETAIL", false);
        params.put("IS_TASK", false);
        params.put("DB_SOURE_USERNAME", kettleConfig.username);
        params.put("DB_SOURE_PASSWORD", kettleConfig.password);
        params.put("DB_SOURE_URL", KettleDbUrlEnum.getEnum(kettleConfig.type).getDesc()
                + kettleConfig.host + ":"
                + kettleConfig.port + ("ORCLE".equals(kettleConfig.type) ? ":" : "/")
                + kettleConfig.name);
        params.put("DB_SOURE_DRIVER_CLASS_NAME", KettleDbDriverEnum.getEnum(kettleConfig.type).getDesc());
        // 添加转换执行记录
        JobRecord jobRecord = new JobRecord();
        if (isTaskDetail) { // 执行的明细作业任务
            params.put("TASK_DETAIL_ID", taskId);
            params.put("IS_TASK_DETAIL", true);
            jobRecord.setFtype(2);
        }
        if (isTask) { // 执行的作业任务
            params.put("TASK_ID", taskId);
            params.put("IS_TASK", true);
            jobRecord.setFtype(1);
        }
        if (!isTask && !isTaskDetail) { //按照分类执行
            params.put("CATEGORY_ID", taskId);
            jobRecord.setFtype(3);
        }
        params.putAll(this.getJobParamMap(listParam)); //处理执行参数
        // 运行状态
        boolean runStatus = true;
        // 执行开始时间
        LocalDateTime startTime = LocalDateTime.now();
        QueryWrapper<JobRecord> wrapper = new QueryWrapper<>();
        wrapper.eq("FTYPE", jobRecord.getFtype());
        wrapper.eq("JOB_ID", taskId);
        log.error(taskId);
        jobRecordService.remove(wrapper);//删除之前的日志只保留最新日志
        jobRecord.setId(DateUtil.generateUUID());
        jobRecord.setJobId(taskId);
        jobRecord.setStatus(RunResultEnum.RUN.getCode());
        jobRecord.setStartTime(startTime);
        jobRecordService.save(jobRecord);
        Job job = this.run(kettleConfig.getKettleRepository(), kettleConfig.jobPath, kettleConfig.jobName
                , null, params
                , LogLevel.getLogLevelForCode(kettleConfig.logLevel));
        // 执行完成后获取日志
        String logText = KettleLogUtil.getLogText(job.getLogChannelId(), true, job.getLogDate().getTime());
        // 判断执行过程中是否有错误
        if (job.getErrors() > 0) {
            runStatus = false;
        }
        LocalDateTime endTime = LocalDateTime.now();
        jobRecord.setStopTime(endTime);   // 执行结束时间
        long numTime = ChronoUnit.SECONDS.between(startTime, LocalDateTime.now()); //所用时间
        jobRecord.setDuration(numTime);
        jobRecord.setLogFile(logText);
        if (runStatus) { //成功执行的，解析日期查询执行的条数
            //"表输入.0 - 完成处理"  "表输出.0 - 完成处理"
            String arr[] = logText.split("\r\n");
            BigDecimal num = BigDecimal.ZERO;
            for (String str : arr) {
                if (str.indexOf("表输出.0 - 完成处理") > -1) {
                    String numStr = str.substring(StringUtil.getIndexOf(str, "W=", 1) + 2, StringUtil.getIndexOf(str, ",", 4));
                    num = num.add(new BigDecimal(numStr));
                    jobRecord.setRowcount(num);
                    continue;
                }
            }
        }
        jobRecord.setStatus(runStatus ? RunResultEnum.SUCCESS.getCode() : RunResultEnum.FAIL.getCode());
        jobRecordService.updateById(jobRecord);
        return true;
    }


    /**
     * 获取参数信息表的值传入kettle
     *
     * @param listParam
     * @return
     */
    private Map<String, Object> getJobParamMap(List<JobParam> listParam) throws Exception {
        Map<String, Object> params = new HashMap<>();
        if (CollectionUtils.isEmpty(listParam)) {
            return null;
        }
        for (JobParam jobParam : listParam) {
            if (null == jobParam || StringUtils.isBlank(jobParam.getKeyName())) {
                continue;
            }
            // 空白的日期类型，年月日。把传过来的日期格式化成年月日
            if (1 == jobParam.getFtype() && 4 == jobParam.getDatatype() && StringUtils.isNotBlank(jobParam.getDefaultValue())) {
                params.put(jobParam.getKeyName(), DateUtil.dateToStrByFormat(DateUtil.strToDate(jobParam.getDefaultValue()), YssEnum.YSS_DEFAULT_DATE_FORMAT.getCode()));
            } else {
                params.put(jobParam.getKeyName(), jobParam.getDefaultValue());
            }
        }
        return params;
    }


    /**
     * 运行资源库中的kjb
     *
     * @param rep          资源库对象
     * @param jobPath      kjb所在路径
     * @param jobName      kjb名称
     * @param versionLabel 版本号，传入null表示执行最新的kjb
     * @param params       kjb需要的命名参数
     * @param logLevel     日志级别
     */
    public Job run(AbstractRepository rep, String jobPath, String jobName, String versionLabel,
                   Map<String, Object> params, LogLevel logLevel) throws Exception {
        // 根据相对目录地址获取ktr所在目录信息
        RepositoryDirectoryInterface rdi = rep.loadRepositoryDirectoryTree().findDirectory(FileUtil.getParentPath(jobPath));
        // 在指定资源库的目录下找到要执行的转换
        JobMeta jm = rep.loadJob(jobName, rdi, new ProgressNullMonitorListener(), versionLabel);
        // 开始执行kjb
        Job job = new Job(rep, jm); // 通过元数据获取kjb的实例
        // 开启进程守护
        job.setDaemon(true);
        // 设置日志级别
        if (logLevel != null) {
            job.setLogLevel(logLevel);
        }
        // 传入kjb需要的变量
        if (CollectionUtil.isNotEmpty(params)) {
            log.info("传入kettle的参数：{}", JsonUtil.toJsonString(params));
            params.forEach((String k, Object v) -> {
                try {
                    job.setParameterValue(k, v.toString());
                    // 设置变量 java代码中变量会覆盖kettle里面设置的变量
                    job.setVariable(k, v.toString());
                } catch (Exception e) {
                    log.error("设置变量信息出错！");
                }
            });
        }
        // 开始执行kjb
        job.start();
        // 线程等待，直到kjb执行完成
        job.waitUntilFinished();
        job.setFinished(true);
        return job;
    }
}
